package com.atguigu.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.UserJumpStats;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import scala.Tuple4;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;

public class UserJumpVisitDim10sApp {
    public static void main(String[] args) throws Exception {
        // create runtime environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // ck

        // 从kafka读取数据
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_visit";

        DataStreamSource<String> pageLogDS = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));

        // 将数据转换成JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjDS = pageLogDS.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts");
                            }
                        }));

        // 按照mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        // 定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String lastPage = value.getJSONObject("page").getString("last_page_id");
                        return lastPage == null || lastPage.length() <= 0;
                    }
                })
                .times(2)
                .consecutive()
                .within(Time.seconds(10));


        // 将模式序列作用于流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        // 获取模式序列过滤出的数据
        OutputTag<JSONObject> timeOut = new OutputTag<JSONObject>("TimeOut") {
        };
        SingleOutputStreamOperator<JSONObject> selectDS = patternStream.select(timeOut,
                new PatternTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                        return map.get("start").get(0);
                    }
                },
                new PatternSelectFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        return map.get("start").get(0);
                    }
                });
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(timeOut);
        DataStream<JSONObject> dataDS = selectDS.union(timeOutDS);

        selectDS.print("SelectDS>>>>>>>>>>>");
        timeOutDS.print("TimeOutDS>>>>>>>>>>>");

        // 将数据转换成javabean
        SingleOutputStreamOperator<UserJumpStats> userJumpStatsDS = dataDS.map(new MapFunction<JSONObject, UserJumpStats>() {
            @Override
            public UserJumpStats map(JSONObject value) throws Exception {
                JSONObject common = value.getJSONObject("common");
                return new UserJumpStats("", "",
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("is_new"),
                        1L,
                        value.getLong("ts"));
            }
        });


        // 分组 聚合 开窗
        SingleOutputStreamOperator<UserJumpStats> userJumpStatsSingleOutputStreamOperator = userJumpStatsDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<UserJumpStats>forBoundedOutOfOrderness(Duration.ofSeconds(12))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserJumpStats>() {
                    @Override
                    public long extractTimestamp(UserJumpStats element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        SingleOutputStreamOperator<UserJumpStats> reduceDS = userJumpStatsSingleOutputStreamOperator.keyBy(new KeySelector<UserJumpStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(UserJumpStats value) throws Exception {
                        return new Tuple4<>(value.getVc(),
                                value.getCh(),
                                value.getAr(),
                                value.getIs_new());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserJumpStats>() {
                            @Override
                            public UserJumpStats reduce(UserJumpStats value1, UserJumpStats value2) throws Exception {
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                return value1;
                            }
                        },
                        new WindowFunction<UserJumpStats, UserJumpStats, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<UserJumpStats> input, Collector<UserJumpStats> out) throws Exception {
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                UserJumpStats next = input.iterator().next();

                                next.setStt(sdf.format(window.getStart()));
                                next.setEdt(sdf.format(window.getEnd()));
                                out.collect(next);
                            }
                        });

        // 写入clickhouse
        reduceDS.print("reduceDS>>>>>>");

        reduceDS.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_userjump_vc_ch_isnew_ar_10s_2022 values(?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
